package com.myorg.util;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.myorg.domain.BlockInfo;

public class DataLocalityRunner {
	public static interface DataLocalityTask {
		public void run(BlockInfo blkInfo, String selectedHost);
	}

	private static String selectHost(BlockInfo blkInfo, Map<BlockInfo, String> map) {
		int score = -1;
		String selected = null;
		try {
			for (String host : blkInfo.getBlockLocation().getHosts()) {
				if (selected == null) {
					selected = host;
					score = getScore(host, map);
				} else {
					int newScore = getScore(host, map);
					if (newScore < score) {
						selected = host;
						score = newScore;
					}
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		}
		return selected;
	}

	private static int getScore(String host, Map<BlockInfo, String> map) {
		int score = 0;
		for (String h : map.values()) {
			if (host.equals(h)) {
				score++;
			}
		}
		return score;
	}

	public static void runWithLocality(List<BlockInfo> blkInfos, DataLocalityTask task) {
		Map<BlockInfo, String> map = new HashMap<BlockInfo, String>();
		if (blkInfos != null) {
			for (BlockInfo blkInfo : blkInfos) {
				String host = selectHost(blkInfo, map);
				map.put(blkInfo, host);
			}

			for (BlockInfo blkInfo : map.keySet()) {
				String host = map.get(blkInfo);
				task.run(blkInfo, host);
			}
		}
	}
}
